package com.sheepone.distributor;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.sheepone.distributor.autoconfigure.WorkerStationConfig;
import com.sheepone.distributor.event.WorkEvent;
import com.sheepone.distributor.exeception.IllegalSizeException;
import com.sheepone.distributor.listener.WorkerEventListener;
import com.sheepone.distributor.thread.SafeThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Miss.杨
 * @description
 * @since 2023-12-07 21:53
 */
public class WorkerStation {
    private final static Logger LOGGER = LoggerFactory.getLogger(WorkerStation.class);
    private final String name;
    private final Disruptor<WorkEvent> workerLine;
    private final ConcurrentHashMap<String, HashSet<WorkerEventListener>> listeners = new ConcurrentHashMap<>();

    public WorkerStation(WorkerStationConfig.Config config) {
        this.name = config.getStation();
        int size = config.getLineSize();
        if (size < 1) {
            throw new IllegalSizeException(name);
        }
        this.workerLine = new Disruptor<>(WorkEvent::new,
                size,
                r -> {
                    Thread thread = new Thread(r);
                    thread.setName(name);
                    thread.setDaemon(true);
                    return thread;
                },
                ProducerType.MULTI,
                new YieldingWaitStrategy()
        );
        SafeThreadPool threadPool = new SafeThreadPool(name, config.getHandlerNum(), config.getKeepAliveTime());
        WorkerStationManager.INSTANCE.registerThreadPool(name, threadPool);
        this.workerLine.handleEventsWith(new WorkerLineHandler(threadPool));
        this.workerLine.start();
        LOGGER.info("【WorkerStation】 \"{}\" start to work", name);
    }

    public boolean register(String topic ,WorkerEventListener listener) {
        HashSet<WorkerEventListener> topicListeners = listeners.get(topic);
        if(topicListeners ==null){
            topicListeners = new HashSet<>(Collections.singleton(listener));
            listeners.put(topic,topicListeners);
            return true;
        }else if(topicListeners.contains(listener)){
            return false;
        }
        topicListeners.add(listener);
        return true;
    }

    public Set<WorkerEventListener> getEventListeners(String topic) {
        return listeners.get(topic);
    }

    /**
     * 暴露发布事件的方法
     */
    public void publish(EventTranslator<WorkEvent> translator) {
        workerLine.getRingBuffer().publishEvent(translator);
    }

    public void stop() {
        workerLine.halt();
        LOGGER.info("【WorkerStation】 \"{}\" has stopped",name);
    }
}
